package com.example.spark.core

import com.example.entity.SogouRecord
import com.example.util.SparkUtil
import com.hankcs.hanlp.HanLP
import com.hankcs.hanlp.seg.common.Term
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

/**
 * @title: SougouSearchLogAnalysis
 * @projectName bigdata
 * @description: SougouSearchLogAnalysis
 *               简单分析搜狗搜素日志
 * @author leali
 * @date 2022/5/15 21:11
 */
object SougouSearchLogAnalysis {
  def SearchLogAnalysis(sparkContext: SparkContext): Unit = {
    val SogouRecordRDD: RDD[SogouRecord] = sparkContext.textFile("src/data/input/SogouQ.sample")
      .map((line: String) => {
        val arr: Array[String] = line.split("\\s+")
        SogouRecord(
          arr(0),
          arr(1),
          arr(2),
          arr(3).toInt,
          arr(4).toInt,
          arr(5)
        )
      })
    // key words
    val wordsRDD: RDD[String] = SogouRecordRDD.flatMap((record: SogouRecord) => {
      //360安全卫士 --> ArrayBuffer(360, 安全卫士)
      //将Java集合转为scala集合
      import scala.collection.JavaConverters._
      HanLP.segment(record.queryWords.replaceAll("[\\[\\]]", "")).asScala.map((_: Term).word)
    })

    // 3.统计指标
    //--1.热门搜索词
    val topHotWord: Array[(String, Int)] = wordsRDD
      .filter((word: String) => !word.equals(".") && !word.equals("+"))
      .map((_: String, 1))
      .reduceByKey((_: Int) + (_: Int))
      .sortBy((_: (String, Int))._2, ascending = false)
      .take(10)
    SparkUtil.printLog("topHotWord")
    topHotWord.foreach(println)

    //--2.用户热门搜索词(带上用户id)
    val userIdAndWord: Array[((String, String), Int)] = SogouRecordRDD.flatMap((record: SogouRecord) => {

      import scala.collection.JavaConverters._
      HanLP.segment(record.queryWords.replaceAll("[\\[\\]]", "")
      ).asScala
        .map((_: Term).word)
        .map((word: String) => (record.userId, word))
    })
      .filter((t: (String, String)) => !t._2.equals(".") && !t._2.equals("+"))
      .map((_: (String, String), 1))
      .reduceByKey((_: Int) + (_: Int))
      .sortBy((_: ((String, String), Int))._2, ascending = false)
      .take(10)
    SparkUtil.printLog("userIdAndWord")
    userIdAndWord.foreach(println)

    //--3.各个时间段搜索热度
    val hotWordByTime: Array[(String, Int)] = SogouRecordRDD.map((record: SogouRecord) => {
      (record.queryTime.substring(0, 5), 1)
    }).reduceByKey((_: Int) + (_: Int))
      .sortBy((_: (String, Int))._2, ascending = false)
      .take(10)
    SparkUtil.printLog("hotWordByTime")
    hotWordByTime.foreach(println)

    sparkContext.stop()
  }
}
